package com.tencent.tbds.demo.spark;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import org.apache.spark.sql.SparkSession;
import scala.Tuple2;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.regex.Pattern;

/**
 * 输入输出准备:
 * vim ~/test1.txt
 * (随意输入单词文本，作为输入文件)
 * 添加如下内容
 * hello spark
 * Hello spark!
 * Hello Spark1
 *
 * `tbdscli getkey`
 *
 * hadoop fs -put ~/test1.txt /tmp/
 *
 *
 * Usage:
 * /usr/hdp/2.2.0.0-2041/spark/bin/spark-submit --master yarn-cluster --executor-memory 3g --driver-memory 1g --num-executors 2 --executor-cores 2 --class com.tencent.tbds.demo.spark.SparkDemo dev-demo-1.0-SNAPSHOT.jar inputFile outputPath
 */

public class SparkDemo {

    private static final Pattern SPACE = Pattern.compile(" ");
    public static final  String fsDefaultFS = "hdfs://hdfsCluster";

    public static void main(String[] args) throws IOException {
        if (args.length < 2) {
            System.err.println("Usage:  /usr/hdp/2.2.0.0-2041/spark/bin/spark-submit " +
                    "--master yarn-cluster --executor-memory 3g --driver-memory 1g --num-executors 2 --executor-cores 2 " +
                    "--class com.tencent.tbds.demo.spark.SparkDemo dev-demo-1.0-SNAPSHOT.jar inputFile outputPath");
            System.exit(1);
        }
        SparkSession spark = SparkSession
                .builder()
                .appName("JavaWordCount")
                .getOrCreate();

        JavaRDD<String> lines = spark.read().textFile(args[0]).javaRDD();

        JavaRDD<String> words = lines.flatMap((FlatMapFunction<String, String>) s -> Arrays.asList(SPACE.split(s)).iterator());

        JavaPairRDD<String, Integer> ones = words.mapToPair(
                (PairFunction<String, String, Integer>) s -> new Tuple2<>(s, 1));

        //如果输出目录存在则删除
        Configuration conf = new Configuration();
        conf.set("fs.defaultFS", fsDefaultFS);
        FileSystem fs = FileSystem.get(conf);
        String outputPath = args[1];
        if(fs.exists(new Path(outputPath))){
            fs.delete(new Path(outputPath), true);
            System.out.println("删除输出目录：" + outputPath + "成功");
        }

        JavaPairRDD<String, Integer> counts = ones.reduceByKey(
                (Function2<Integer, Integer, Integer>) Integer::sum);
        counts.saveAsTextFile(args[1]);

        List<Tuple2<String, Integer>> output = counts.collect();
        for (Tuple2<String, Integer> tuple : output) {
            System.out.println(tuple._1() + ": " + tuple._2());
        }
        spark.stop();
    }

}
